package com.pangyangyang.elasticsearch.plugin.jdbc.pool;

import com.pangyangyang.elasticsearch.plugin.exception.MessageRiverException;
import com.pangyangyang.elasticsearch.plugin.jdbc.constant.JDBCHandlerConstants;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiverSettings;
import org.apache.commons.lang.NotImplementedException;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;

import javax.sql.DataSource;
import java.io.PrintWriter;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

/**
 * Created by yangwang on 2015/1/20.
 */
public class JDBCConnectionPool implements DataSource{

    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);
    //metadata
    private String url, username, password;
    private int loginTimeout = DEFAULT_LOGIN_TIMEOUT;

    //pool config
    private int maxSize, minSize;
    private AtomicInteger ctl;

    private Set<Connection> availableConnectionSet;
    private Set<ConnectionLease> takenConnectionSet;
    private long maxLeaseMillis;
    private Lock mainLock;

    public static final int DEFAULT_LOGIN_TIMEOUT = 20;
    public static final int DEFAULT_MAX_LEASE_SECONDS = 60;
    protected static final Class<Connection>[] CONNECTION_INTERFACES = new Class[]{Connection.class};

    public JDBCConnectionPool(String url, String username, String password) {
        this.url = url;
        this.username = username;
        this.password = password;
        maxLeaseMillis = TimeUnit.SECONDS.toMillis(DEFAULT_MAX_LEASE_SECONDS);
        availableConnectionSet = new HashSet<>();
        takenConnectionSet = new HashSet<>();
        this.mainLock = new ReentrantLock();
    }

    public Connection getConnection() throws SQLException{

        return establishConnection();
    }

    public void shutdown(){

    }

    protected Connection establishConnection() throws SQLException{
        Connection connection = null ;
        DriverManager.setLoginTimeout(getLoginTimeout());
        connection = DriverManager.getConnection(url, username, password);
        connection.setClientInfo(JDBCHandlerConstants.SETTING_KEY_URL, url);

        Connection proxyConnection =
                (Connection)Proxy.newProxyInstance(connection.getClass().getClassLoader(),
                        CONNECTION_INTERFACES, new JDBCConnectionProxy(connection, this));

        return proxyConnection;
    }

    protected List<Connection> clearExpireConnections(){
        List<Connection> expiredConnections = new ArrayList<>();
        mainLock.lock();
        for(ConnectionLease connectionLease : takenConnectionSet){
            if (connectionLease.isExpire()){
                takenConnectionSet.remove(connectionLease.getConnection());
                expiredConnections.add(connectionLease.getConnection());
            }
        }
        mainLock.unlock();

        for (Connection expiredConnection : expiredConnections){
            closeConnection(expiredConnection);
        }

        return expiredConnections;
    }

    public void addConnection(Connection connection){
        mainLock.lock();
        availableConnectionSet.add(connection);
        mainLock.unlock();
    }

    public void removeConnection(Connection connection){
        mainLock.lock();
        try {
            if (availableConnectionSet.remove(connection)){
                return;
            }
            takenConnectionSet.remove(connection);
        } finally {
            mainLock.unlock();
        }
    }

    public void setMaxLeaseSeconds(int maxLeaseSeconds) {
        this.maxLeaseMillis = TimeUnit.SECONDS.toMillis(maxLeaseSeconds);
    }

    /**
     * 释放Connection
     * @param connection
     * @return 成功释放true，否则false。释放失败应该及时关闭Connection
     */
    public boolean releaseConnection(Connection connection){
        mainLock.lock();
        try {
            if (takenConnectionSet.remove(connection)){
                availableConnectionSet.add(connection);
                return true;
            }
            return false;
        } finally {
            mainLock.unlock();
        }
    }

    public void closeAllConnection(){
        for(;;){
            mainLock.lock();
            if(availableConnectionSet.isEmpty() && takenConnectionSet.isEmpty()){
                break;
            }
            Iterator<Connection> connectionIterator = availableConnectionSet.iterator();
            for(; connectionIterator.hasNext();){

            }
            mainLock.unlock();
        }
    }

    private boolean closeConnection(Connection connection){
        JDBCConnectionProxy connectionProxy =
                (JDBCConnectionProxy) connection;
        Connection realConnection = connectionProxy.getConnection();
        try {
            if (realConnection.isClosed()) {
                return true;
            }
            realConnection.close();
        } catch (SQLException ignore){
            return false;
        }

        return true;
    }

    /**
     * 记录Connection的租用信息，用于关闭超期不归还的Connection。
     */
    private class ConnectionLease{
        private long deadline;
        private Connection connection;

        public ConnectionLease(Connection connection) {
            this.deadline = System.currentTimeMillis() + maxLeaseMillis;
            this.connection = connection;
        }

        public boolean isExpire(){
            return (deadline - System.currentTimeMillis()) <= 0;
        }

        public Connection getConnection() {
            return connection;
        }
    }

    @Override
    public Connection getConnection(String username, String password) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public PrintWriter getLogWriter() throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setLogWriter(PrintWriter out) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public int getLoginTimeout() {
        return loginTimeout;
    }

    @Override
    public void setLoginTimeout(int loginTimeout) {
        this.loginTimeout = loginTimeout;
    }

    @Override
    public Logger getParentLogger() throws SQLFeatureNotSupportedException {
        throw new UnsupportedOperationException();
    }

    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isWrapperFor(Class<?> iface) throws SQLException {
        throw new UnsupportedOperationException();
    }
}
